package com.boot.netty.demo1;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.util.CharsetUtil;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;

public class SimpleNettyServer {

    private static final int PORT = 8080;
    private static final int MAX_MESSAGE_SIZE = 1024; // 1KB消息大小
    private static final AtomicInteger messageCount = new AtomicInteger(0);
    private static final ConcurrentLinkedQueue<Channel> channels = new ConcurrentLinkedQueue<>();

    public static void main(String[] args) throws InterruptedException, IOException {
        // 创建EventLoopGroup对象，用于处理网络事件
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            // 创建ServerBootstrap对象
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            // 设置ChannelPipeline中的处理器
//                            ch.pipeline().addLast(new FixedLengthFrameDecoder(MAX_MESSAGE_SIZE));
                            ch.pipeline().addLast(new StringDecoder(CharsetUtil.UTF_8));
                            ch.pipeline().addLast(new SimpleChannelInboundHandler<String>() {
                                @Override
                                protected void channelRead0(ChannelHandlerContext ctx, String msg) {
                                    System.out.println("Received message: " + msg);
                                    // 每当接收到消息时，计数器增加
                                    messageCount.incrementAndGet();
                                }

                                @Override
                                public void channelActive(ChannelHandlerContext ctx) {
                                    channels.add(ctx.channel());
                                }

                                @Override
                                public void channelInactive(ChannelHandlerContext ctx) {
                                    channels.remove(ctx.channel());
                                }
                            });
                        }
                    })
                    .option(ChannelOption.SO_BACKLOG, 128)
                    .childOption(ChannelOption.SO_KEEPALIVE, true);

            // 绑定端口，开始接收进来的连接
            ChannelFuture f = b.bind(PORT).sync();

            // 新建一个线程用于读取控制台的输入并发送给所有客户端
            new Thread(() -> {
                BufferedReader in = new BufferedReader(new InputStreamReader(System.in));
                while (true) {
                    String line;
                    try {
                        line = in.readLine();
                    } catch (IOException e) {
                        e.printStackTrace();
                        continue;
                    }

                    for (Channel channel : channels) {
                        channel.writeAndFlush(line + "\r\n");
                    }
                }
            }).start();

            // 等待服务器socket关闭
            f.channel().closeFuture().sync();
        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }

        // 打印出总共接收到的消息数量
        System.out.println("Total messages received: " + messageCount.get());
    }
}